package com.nedvisol.hbaserest.server;

import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.InputStreamReader;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.NavigableMap;

import javax.ws.rs.Consumes;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;

import org.apache.commons.collections.map.LinkedMap;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.conf.Configuration;

import com.nedvisol.hbaserest.data.Example;
import com.nedvisol.hbaserest.data.JaxbList;
import com.nedvisol.hbaserest.request.ColumnData;
import com.nedvisol.hbaserest.request.MultiRowData;
import com.nedvisol.hbaserest.request.RowData;
import com.nedvisol.hbaserest.request.TableSchema;

@Path("/")
@Produces("application/json")
public class HBaseRestServiceImpl implements HBaseRestService {

	private String zookeeperQuorum;

	public String getZookeeperQuorum() {
		return zookeeperQuorum;
	}

	public void setZookeeperQuorum(String zookeeperQuorum) {
		this.zookeeperQuorum = zookeeperQuorum;
	}

	public String getZookeeperPort() {
		return zookeeperPort;
	}

	public void setZookeeperPort(String zookeeperPort) {
		this.zookeeperPort = zookeeperPort;
	}

	private String zookeeperPort;

	private Configuration conf = null;

	private Configuration getConfiguration() {
		if (conf == null) {
			Configuration c = new Configuration();
			c.set("hbase.zookeeper.quorum", this.zookeeperQuorum);
			c.set("hbase.zookeeper.property.clientPort", this.zookeeperPort);
			this.conf = c;
		}
		return this.conf;
	}

	@PUT
	@Path("/table/{tableName}/")
	@Produces("application/json")
	public String createTable(@PathParam("tableName") String tableName,
			TableSchema tableSchema) throws Exception {
		Configuration conf = this.getConfiguration();
		HBaseAdmin admin = new HBaseAdmin(conf);
		HTableDescriptor desc = new HTableDescriptor(tableName);
		Collection<String> columnFamilies = tableSchema.columnFamilies;
		for (String cf : columnFamilies) {
			desc.addFamily(new HColumnDescriptor(cf));
		}
		admin.createTable(desc);

		return tableName;
	}

	@PUT
	@Path("/{tableName}/new-rows")
	@Produces("application/json")
	public int insertUpdateRows(@PathParam("tableName") String tableName,
			MultiRowData rowsData) throws Exception {
		Configuration conf = this.getConfiguration();
		HTable table = new HTable(conf, tableName);
		for (RowData rowData : rowsData.rows) {
			Put row = new Put(rowData.key);
			for (Map.Entry<String, byte[]> e : rowData.columns.entrySet()) {
				String column = e.getKey();
				String[] cf = column.split(":");
				row.add(Bytes.toBytes(cf[0]), Bytes.toBytes(cf[1]),
						e.getValue());
			}
			if (rowData.checkOption != null) {
				String colName = rowData.checkOption.columnName;
				String[] cf = colName.split(":");
				boolean updated = table.checkAndPut(rowData.key,
						Bytes.toBytes(cf[0]), Bytes.toBytes(cf[1]),
						rowData.checkOption.value, row);
				return updated ? 1 : 0;

			} else {
				table.put(row);
			}
		}
		return rowsData.rows.size();
	}

	private RowData[] getFromScan(String tableName, String rowKey)
			throws Exception {
		Configuration conf = this.getConfiguration();
		HTable table = new HTable(conf, tableName);
		Scan scanner = new Scan();
		rowKey = rowKey.replaceAll("\\*", "");
		scanner.setStartRow(Bytes.toBytes(rowKey));
		ResultScanner rsScan = table.getScanner(scanner);
		Result res = null;
		ArrayList<RowData> resultList = new ArrayList<RowData>();
		while ((res = rsScan.next()) != null) {
			String key = Bytes.toString(res.getRow());
			if (!key.startsWith(rowKey)) {
				break;
			}			
			
			RowData rd = new RowData();
			rd.key = res.getRow();

			NavigableMap<byte[], NavigableMap<byte[], byte[]>> columns = res
					.getNoVersionMap();
			LinkedHashMap<String, byte[]> colMap = new LinkedHashMap<String, byte[]>();
			for (Map.Entry<byte[], NavigableMap<byte[], byte[]>> cf : columns
					.entrySet()) {
				String colFam = Bytes.toString(cf.getKey());
				for (Map.Entry<byte[], byte[]> col : cf.getValue().entrySet()) {
					String colName = colFam + ":"
							+ Bytes.toString(col.getKey());
					colMap.put(colName, col.getValue());
				}
			}
			rd.columns = colMap;
			resultList.add(rd);
		}
		return resultList.toArray(new RowData[0]);
	}

	@GET
	@Path("/{tableName}/{rowKey}/")
	@Produces("application/json")
	public RowData[] getRow(@PathParam("tableName") String tableName,
			@PathParam("rowKey") String rowKey) throws Exception {
		if (rowKey.endsWith("*")) {
			return this.getFromScan(tableName, rowKey);
		}
		Configuration conf = this.getConfiguration();
		HTable table = new HTable(conf, tableName);
		Get row = new Get(Bytes.toBytes(rowKey));
		/*
		 * HTableDescriptor desc = table.getTableDescriptor();
		 * Collection<HColumnDescriptor> cfs = desc.getFamilies();
		 * for(HColumnDescriptor col : cfs) { row.addFamily(col.getName()); }
		 */

		Result res = table.get(row);

		if (res.isEmpty()) {
			return null;
		}

		RowData rd = new RowData();
		rd.key = res.getRow();

		NavigableMap<byte[], NavigableMap<byte[], byte[]>> columns = res
				.getNoVersionMap();
		LinkedHashMap<String, byte[]> colMap = new LinkedHashMap<String, byte[]>();
		for (Map.Entry<byte[], NavigableMap<byte[], byte[]>> cf : columns
				.entrySet()) {
			String colFam = Bytes.toString(cf.getKey());
			for (Map.Entry<byte[], byte[]> col : cf.getValue().entrySet()) {
				String colName = colFam + ":" + Bytes.toString(col.getKey());
				colMap.put(colName, col.getValue());
			}
		}
		rd.columns = colMap;
		RowData[] rdArray = new RowData[1];
		rdArray[0] = rd;
		return rdArray;
	}

	@POST
	@Path("/jsontest/")
	@Produces("application/json")
	public MultiRowData jsonTest(LinkedHashMap<String, Object> map) {
		ArrayList<String> a = new ArrayList<String>();
		a.add((String) map.get("test"));
		a.add("test2");
		JaxbList jl = new JaxbList(a);

		MultiRowData rd = new MultiRowData();
		rd.rows = new ArrayList<RowData>();
		RowData r = new RowData();
		r.key = Bytes.toBytes("rowid");
		r.columns = new LinkedHashMap<String, byte[]>();
		r.columns.put("test", Bytes.toBytes("hahaha"));
		rd.rows.add(r);
		return rd;
	}

}
